Airflow 3.0: Datasets Become Assets

The Rename, in Code

# Airflow 2.x
from airflow.datasets import Dataset
orders_dataset = Dataset("s3://data-lake/orders/daily.csv")

# Airflow 3.x
from airflow.sdk import Asset
orders_asset = Asset("s3://data-lake/orders/daily.csv")

Everything from the Datasets page carries over unchanged conceptually — outlets=[orders_asset] on a producing task, schedule=[orders_asset] on a consuming DAG. Airflow 3 keeps a compatibility shim so Dataset still works as an alias, but new code should use Asset.

What's Actually New: AssetWatchers

In 2.x, a Dataset only updates when an Airflow task explicitly declares it via outlets=. In 3.0, an AssetWatcher can trigger on changes happening outside Airflow entirely — a message landing in a queue, a webhook firing, a file appearing — without any Airflow task needing to run first:

from airflow.sdk import Asset, AssetWatcher
from airflow.providers.amazon.aws.triggers.sqs import SqsSensorTrigger

# Watches an SQS queue directly - no Airflow task publishes this Asset,
# the queue message itself is the update signal
trigger = SqsSensorTrigger(
    sqs_queue="https://sqs.us-east-1.amazonaws.com/123456789012/orders-updated",
)

orders_asset = Asset(
    "s3://data-lake/orders/daily.csv",
    watchers=[AssetWatcher(name="orders-queue-watcher", trigger=trigger)],
)

@dag(schedule=[orders_asset], ...)
def consume_orders_asset():
    ...

This closes a real gap from the Datasets model: previously, if the file landed in S3 via some other process (not an Airflow DAG), there was no clean way to make Airflow react to it without an Airflow task in the loop somewhere. AssetWatchers let the Asset itself watch the outside world.

Asset Aliases and Multi-Asset Outlets

Airflow 3 also allows a task to declare which Assets it updates dynamically, at runtime, rather than only statically in the decorator:

from airflow.sdk import Asset, task

@task(outlets=[Asset("s3://data-lake/orders/daily.csv")])
def extract_and_load(**context):
    # ... does the actual work ...
    # Can also yield additional runtime-determined Asset events:
    from airflow.sdk import AssetEvent
    yield AssetEvent(uri="s3://data-lake/orders/daily.csv", extra={"row_count": 4821})

The extra payload travels with the Asset event and is readable by downstream consumers — useful for passing lightweight metadata (a row count, a data quality flag) alongside the "this changed" signal itself.

This content is reserved for Premium Members.